package com.pw.study.flink.source

import java.util.Random

import com.pw.study.flink.entities.WordBeanOne
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._

class CustomWordSource extends RichSourceFunction[WordBeanOne] {
  private var isSwitch = true

  override def run(ctx: SourceFunction.SourceContext[WordBeanOne]): Unit = {
    val line = "Our diurnal existence is divided into two phases  as distinct as day and night. We call them work and play. We work so many hours a day. And  when we have allowed the necessary minimum for such activities as eating and shopping  the rest we spend in various activities which are known as recreations  an elegant word which disguises the fact that we usually do not even play in our hours of leisure  but spend them in various forms of passive enjoyment or entertainment―not playing football but watching football matches not acting but theatre-going Not walking but riding in a motor coach We need to make  therefore  a hard-and-fast distinction not only between work and play but equally between active play and passive entertainment It is  I suppose  the decline of active play―of amateur sport―and the enormous growth of purely receptive entertainment which has given rise to a sociological interest in the problem. If the greater part of the population  instead of indulging in sport  spend their hours of leisure viewing television programmers  there will inevitably be a decline in health and physique"
    val arr: Array[String] = line.split(" ")
    val random = new Random()
    while(isSwitch){
      val len: Int = random.nextInt(arr.length)

      val bean = WordBeanOne(arr(len), 1L)
      ctx.collect(bean)
      println("产生数据： "+bean)
      Thread.sleep(200)
    }
  }

  override def cancel(): Unit = {
    isSwitch = false
  }
}
